package bigbeard.tools.net.server.impl;


import bigbeard.tools.net.api.StateChangeCallback;
import bigbeard.tools.net.api.server.IConnector;
import bigbeard.tools.net.server.BaseServerImpl;
import bigbeard.tools.thread.pools.ThreadPoolFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;

/**
 * @author bigbeard on 2022-09-29
 */
public class TcpSocketServerImpl extends BaseServerImpl {

    private static final int READ_BUFF_SIZE = 8192;
    private static final int CLIENT_CLOSE = -1;
    private static final String THREAD_CONNECT_NAME_PREFIX = "thread-server-net-main-%s";

    private volatile boolean isRun;

    private int listenPort;

    private void execute() throws IOException {
        if (isRun) {
            return;
        }
        isRun = true;
        // 创建选择器
        Selector selector = Selector.open();
        // 打开监听信道
        ServerSocketChannel listenerChannel = ServerSocketChannel.open();
        listenerChannel.socket().bind(new InetSocketAddress(listenPort));
        // 设置为非阻塞模式
        listenerChannel.configureBlocking(false);
        // 将选择器绑定到监听信道,只有非阻塞信道才可以注册选择器.并在注册过程中指出该信道可以进行Accept操作
        listenerChannel.register(selector, SelectionKey.OP_ACCEPT);
        notifyState("", StateChangeCallback.SERVER_OPEN, "服务器在[" + listenPort + "]启动成功");
        while (isRun) {
            int readyChannels = selector.select();
            if (readyChannels == 0) {
                continue;
            }
            //返回此选择器的已选择键集
            Set<SelectionKey> keys = selector.selectedKeys();
            Iterator<SelectionKey> iterKey = keys.iterator();

            while (iterKey.hasNext() && isRun) {
                SelectionKey sk = iterKey.next();
                // 移除处理过的键
                iterKey.remove();
                if (sk.isAcceptable() && isRun) {
                    acceptClient(sk);
                }
                if (sk.isReadable() && isRun) {
                    readData(sk);
                }
            }
        }
        selector.close();
        notifyState("", StateChangeCallback.SERVER_CLOSE, "服务器已经关闭");
    }

    private void acceptClient(SelectionKey key) throws IOException {
        if (null == key) {
            return;
        }
        SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
        // 非阻塞式
        clientChannel.configureBlocking(false);

        String ipAndPort = clientChannel.socket().getInetAddress()
                           + ":" + clientChannel.socket().getPort();
        // 注册到selector
        clientChannel.register(key.selector(), SelectionKey.OP_READ, ipAndPort);
        key.interestOps(SelectionKey.OP_ACCEPT);
        IConnector connector = new TcpSocketConnectorImpl();
        connector.setConnector(clientChannel);
        connector.setCode(ipAndPort);
        connector.updateLastMsgTime();
        notifyClientAccept(ipAndPort, connector);
    }


    private void readData(SelectionKey selectionKey) {
        if (null == selectionKey) {
            return;
        }
        SocketChannel clientChannel = (SocketChannel) selectionKey.channel();
        String key = selectionKey.attachment().toString();

        ByteBuffer buffer = ByteBuffer.allocate(READ_BUFF_SIZE);

        // 读取信息获得读取的字节数
        int bytesRead;
        try {
            while ((bytesRead = clientChannel.read(buffer)) > 0) {
                buffer.flip();
                byte[] b = new byte[bytesRead];
                buffer.get(b, 0, b.length);

                notifyDataArrival(key, b);
                buffer.clear();
            }
            if (bytesRead == CLIENT_CLOSE) {
                notifyState(key, StateChangeCallback.CLIENT_CLOSE, "客户端主动关闭");
                closeSelectionKey(selectionKey);
            }
        } catch (IOException e) {
            selectionKey.cancel();
            closeSelectionKey(selectionKey);
            notifyState(key, StateChangeCallback.CLIENT_CLOSE, "客户端异常,导致连接关闭");
            e.printStackTrace();
        }
    }

    private void closeSelectionKey(SelectionKey selectionKey) {
        if (selectionKey != null && selectionKey.channel() != null) {
            try {
                selectionKey.channel().close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }

    @Override
    public void start(int port) throws Exception {
        listenPort = port;
        ExecutorService executorServiceConnect = ThreadPoolFactory.createSingleThreadPool(THREAD_CONNECT_NAME_PREFIX);
        executorServiceConnect.execute(() -> {
            try {
                isRun = false;
                execute();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    @Override
    public void stop() {
        isRun = false;
    }
}
